package com.ch1.kafka001.hello_kafka;

import com.ch1.kafka001.constant.KafkaInfo;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class HelloKafkaConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", KafkaInfo.HOST);
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer",StringDeserializer.class);

        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);

        try{

            consumer.subscribe(Collections.singletonList("test2"));
            while(true){
                ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(500));
                for(ConsumerRecord<String,String> record: records){
                    System.out.println("topic:"+record.topic()+"  分区："+record.partition()+"  偏移量："
                            +record.offset()+"  key:"+record.key()+"  value:"+record.value());
                }
            }

        }finally {
            consumer.close();
        }
    }
}




































